package com.lovecaa;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 异步发送消息
 * 创建时间： 2023年10月20日 23:29:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class AsyncProducer {
    public static void main(String[] args) {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("AsyncProducerGroup");
            producer.setNamesrvAddr("192.168.192.110:9876");
            producer.start();
            // 构造消息发送条数
            int msgSize = 100;
            CountDownLatch countDownLatch = new CountDownLatch(msgSize);
            for (int i = 0; i < msgSize; i++) {
                String topic = "TopicTest";
                String tag = "TagA";
                byte[] msg = ("AsyncProducer_" + i).getBytes(StandardCharsets.UTF_8);
                Message message = new Message(topic, tag, msg);
                // 异步发送消息使用callback进行回调
                final int index = i;
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.println("消息" + index + "发送成功:" + sendResult);
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        countDownLatch.countDown();
                        System.out.println("消息" + index + "发送失败:" + throwable.getMessage());
                    }
                });
            }
            // 等待消息发送完成后再关闭生产者
            boolean await = countDownLatch.await(10, TimeUnit.SECONDS);
            System.out.println("countDownLatch：" + await);
            producer.shutdown();
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }
}
